package jupitermouse.site.example.project

import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests
import org.slf4j.LoggerFactory

import scala.collection.mutable.ArrayBuffer

object LogAnalysis {
  //生产上记录日志采用这种方式
  val logger = LoggerFactory.getLogger("LogAnalysis")

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //接受Kafka数据
    val topic = "flink"

    val properties = new Properties();
    properties.setProperty("bootstrap.servers", "192.168.81.130:9092")
    properties.setProperty("zookeeper.connect", "192.168.81.130:2181")
    properties.setProperty("group.id","test-flink")
    val consumer = new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(),properties)

    import org.apache.flink.api.scala._

    val data = env.addSource(consumer)

//    data.print().setParallelism(1)

    //数据处理
    val logdata = data.map(x => {
      val split = x.split("\t")
      val level = split(2)
      val timeStr = split(3)
      val domain = split(5)
      val traffic = split(6).toLong
      logger.info(""+(level,transfromTime(timeStr),domain,traffic))
      (level,transfromTime(timeStr),domain,traffic)
    }).filter(_._2 != 0).filter(_._1 == "E")
        .map(x => (x._2,x._3,x._4)) // 1 level(抛弃)  2 time  3 domain   4 traffic
    logdata.assignTimestampsAndWatermarks(
      new AssignerWithPeriodicWatermarks[(Long,String,Long)] {
        val maxOutOfOrderness = 3500 // 3.5 seconds 最大无序容忍时间

        var currentMaxTimestamp: Long = _

        override def getCurrentWatermark: Watermark = {
          new Watermark(currentMaxTimestamp - maxOutOfOrderness)
        }

        override def extractTimestamp(element: (Long, String, Long), previousElementTimestamp: Long): Long = {
          val timestamp = element._1
          currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
          timestamp
        }
      }

    ).keyBy(1) //按照域名进行keyBy
        .window(TumblingEventTimeWindows.of(Time.seconds(60)))
        .apply(new WindowFunction[(Long,String,Long),(String,String,Long),Tuple,TimeWindow] {
          override def apply(key: Tuple, window: TimeWindow, input: Iterable[(Long, String, Long)], out: Collector[(String, String, Long)]): Unit = {

            val domain = key.getField(0).toString
            var sum = 0l
            val times = ArrayBuffer[Long]()

            val iterator = input.iterator
            while (iterator.hasNext){
              val next = iterator.next()
              sum += next._3 //traffic求和
              //TODO 能拿到windows里面的时间
              times.append(next._1)
            }
            val time = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date(times.max))
            out.collect((time,domain,sum)) //时间，域名，流量的和
          }
        }).print().setParallelism(1)

    env.execute("LogAnalysis")

  }

  def write2ES(input:DataStream[(String, String, Long)]): Unit ={
    val httpHosts = new java.util.ArrayList[HttpHost]
    httpHosts.add(new HttpHost("192.168.81.130", 9200, "http"))

    import org.apache.flink.api.scala._

    val esSinkBuilder = new ElasticsearchSink.Builder[(String,String,Long)](
      httpHosts,
      new ElasticsearchSinkFunction[(String,String,Long)] {
        def createIndexRequest(element: (String,String,Long)): IndexRequest = {
          val json = new java.util.HashMap[String, Any]
          json.put("time", element._1)
          json.put("domain", element._2)
          json.put("traffics", element._3)

          val id = element._1 + "-" + element._2
          return Requests.indexRequest()
            .index("cdn")
            .`type`("traffic")
            .id(id)
            .source(json)
        }

        override def process(element: (String, String, Long), ctx: RuntimeContext, indexer: RequestIndexer): Unit = {
          indexer.add(createIndexRequest(element))
        }
      }

    )
    // configuration for the bulk requests; this instructs the sink to emit after every element, otherwise they would be buffered
    esSinkBuilder.setBulkFlushMaxActions(1)
    // provide a RestClientFactory for custom configuration on the internally created REST client
/*    esSinkBuilder.setRestClientFactory(
      restClientBuilder -> {
        restClientBuilder.setDefaultHeaders(...)
        restClientBuilder.setMaxRetryTimeoutMillis(...)
        restClientBuilder.setPathPrefix(...)
        restClientBuilder.setHttpClientConfigCallback(...)
      }
    )*/

    // finally, build and add the sink to the job's pipeline
    input.addSink(esSinkBuilder.build)
  }

  /**
    * 转换时间格式，注意数据不符合，那么过滤掉
    * @param timeStr 时间字符串
    * @return Long
    */
  def transfromTime(timeStr:String): Long ={
    try {
      val sourceFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      val time = sourceFormat.parse(timeStr).getTime
      return time
    } catch {
      case e: Exception => {
        logger.error(s"time parse error: $timeStr", e.getMessage)
      }
    }
    return 0l
  }

}
